package com.jd.rapid.rpc.server;

import com.jd.rapid.rpc.client.RpcClientHandler;
import com.jd.rapid.rpc.codec.RpcDecoder;
import com.jd.rapid.rpc.codec.RpcEncode;
import com.jd.rapid.rpc.codec.RpcRequest;
import com.jd.rapid.rpc.codec.RpcResponse;
import com.jd.rapid.rpc.config.provider.ProviderConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;


@Slf4j
public class RpcServer {

    private String serverAddress;

    private EventLoopGroup bossGroup=new NioEventLoopGroup();

    private EventLoopGroup workerGroup=new NioEventLoopGroup();

    private volatile Map<String /**interface**/,Object> handlerMap=new HashMap<>();

    public RpcServer(String serverAddress) throws InterruptedException {
        this.serverAddress=serverAddress;
        this.start();
    }

    private void start() throws InterruptedException {
        ServerBootstrap serverBootstrap=new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                //tcp=sync+accept=backlog
                .option(ChannelOption.SO_BACKLOG,1024)
                //添加控制台日志输出
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //编解码的handler
                        ChannelPipeline cp=socketChannel.pipeline();
                        //0-4是包头，解码辅助
                        cp.addLast(new LengthFieldBasedFrameDecoder(65535,0,4,0,0));
                        //server端对request解码
                        cp.addLast(new RpcDecoder(RpcRequest.class));
                        //server端对response编码
                        cp.addLast(new RpcEncode(RpcResponse.class));
                        //实际的业务处理器
                        cp.addLast(new RpcServerHandler(handlerMap));
                    }
                });
          String[]  array= serverAddress.split(":");
          String host=array[0];
          int port=Integer.parseInt(array[1]);
          ChannelFuture channelFuture=serverBootstrap.bind(host,port).sync();
          channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                       if(channelFuture.isSuccess()){
                           log.info("server scuccess bing to :"+serverAddress);
                       }else{
                           log.error("server fail bing to "+serverAddress);
                           throw  new Exception("server start fail ,cause:"+channelFuture.cause());
                       }
                }
          });
        try {
            channelFuture.await(5000, TimeUnit.MILLISECONDS);
            if(channelFuture.isSuccess()){
                log.info("start rapid rpc successful");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.error("start rapid rpc occur Interrupted,ex： "+e);
        }
    }


    /**
     * 程序注册器
     */
    public void registerProccessor(ProviderConfig providerConfig){
            //key: providerConfig.interface （userService接口权限命名）
            //value: providerConfig.ref (userService接口下具体的实现类userServiceImpl实例对象)
            handlerMap.put(providerConfig.getInterface(),providerConfig.getRef());
    }

    public void close(){
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
    }
}
